package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.RedisCacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

/**
 * @author DiXiaoNie
 * @since 2022/12/24 20:21
 */
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private RedisCacheService redisCacheService;

    @Resource
    private ThreadPoolTaskExecutor taskExecutor;

    /**
     * 重新加载数据
     *
     * @author DiXiaoNie
     * @date 2022/12/25
     */
    @Override
    @Scheduled(cron = "0 */10 * * * ?")
    public void reloadData() {
        log.info("数据库数据同步到缓存");

        //1.清除缓存中的数据，是有争议，我们暂时先不清除
        Set<String> listKeys = this.redisCacheService.scan("topic_*");
        Set<String> zSetKeys = this.redisCacheService.scan("future_*");
        this.redisCacheService.delete(listKeys);
        this.redisCacheService.delete(zSetKeys);

        //2.查询数据库表中taskinfo，执行时间小于等于当前时间+五分钟的数据
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        LambdaQueryWrapper<Taskinfo> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.le(Taskinfo::getExecuteTime, calendar.getTimeInMillis());
        List<Taskinfo> taskinfos = this.taskinfoMapper.selectList(queryWrapper);
        //3.保存数据到redis中，有可能到list，也有可能到zset中
        for (Taskinfo taskinfo : taskinfos) {
            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(taskinfo.getExecuteTime().getTime());
            this.saveTaskToRedis(task);
        }
    }

    /**
     * 刷新
     *
     * @author DiXiaoNie
     * @date 2022/12/25
     * 加入定时器
     * 注意：当前这个定时任务不会被任何接口调用，自己完成定时任务。所以此方法没有参数
     * 第二：由于是定时任务 方法同样也不会有任何返回值
     */
    @Override
    //@Scheduled(cron = "0 */1 * * * ?")
    @Scheduled(cron = "*/1 * * * * ?") //换成每秒中执行一次，方便测试
    public void refresh() {
        //String token = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        //if (StringUtils.isNotBlank(token)) {

        try {
            //加入休眠时间
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
        //1.获取zset中的所有key,判断当前数据是否到期
        Set<String> zSetKeys = this.redisCacheService.scan("future_*");
        for (String zSetKey : zSetKeys) {
            //根据score值（执行时间）小于等于当前系统时间，则取出来 taskZSet表示到期数据
            Set<String> taskZSet = this.redisCacheService.zRangeByScore(zSetKey, 0, System.currentTimeMillis());
            //2.如果到期，则提取出数据，并删除数据，然后同步到list队列中
            if (taskZSet != null && taskZSet.size() > 0) {
                //同步数据
                String future_key = zSetKey;
                String topic_key = ScheduleConstants.TOPIC + zSetKey.split(ScheduleConstants.FUTURE)[1];
                this.redisCacheService.refreshWithPipeline(future_key, topic_key, taskZSet);
                System.out.println("成功的将" + future_key + "下的当前需要执行的任务数据刷新到" + topic_key + "下");
            }
        }
        //}


    }

    /**
     * 拉取任务
     *
     * @param type     类型
     * @param priority 优先级
     * @author DiXiaoNie
     * @date 2022/12/24
     */
    @Override
    public Task pullTask(Integer type, Integer priority) {
        //使用多线程的方式
        Future<Task> future = this.taskExecutor.submit(new Callable<Task>() {
            @Override
            public Task call() throws Exception {
                String keySuffix = type + "_" + priority;
                //1.从redis的list中取出数据并删除
                String taskStr = redisCacheService.lRightPop(ScheduleConstants.TOPIC + keySuffix);
                if (taskStr != null) {
                    //3.把获取到taskStr转化成对象类型，从而获取taskId参数
                    Task task = JSON.parseObject(taskStr, Task.class);
                    //2.修改数据库的表，删除任务表以及修改任务日志表的状态以及version
                    taskinfoMapper.deleteById(task.getTaskId());
                    TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
                    if (taskinfoLogs != null) {
                        taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);//表示消费状态
                        //按照正常来说我们还要对version更新,此时不需要，因为mp已经集成version乐观锁
                        //taskinfoLogs.setVersion();
                        taskinfoLogsMapper.updateById(taskinfoLogs);
                    }
                    //4.返回数据
                    return task;
                }
                return null;
            }
        });
        try {
            return future.get();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }

    }

    /**
     * 添加任务
     *
     * @param task 任务对象
     * @author DiXiaoNie
     * @date 2022/12/24
     */
    @Override
    public void addTask(Task task) {
        taskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //1.新增任务到数据库表中，taskinfo和taskinfo_logs
                Boolean flag = saveTaskToDB(task);
                if (!flag) {
                    return;
                }
                //2.添加任务数据到redis中
                saveTaskToRedis(task);
            }
        });
    }

    /**
     * 保存任务到redis,
     *
     * @param task 任务
     * @author DiXiaoNie
     * @date 2022/12/24
     */
    private void saveTaskToRedis(Task task) {
        //定义一个key前缀
        String keySuffix = task.getTaskType() + "_" + task.getPriority();
        //2.1 当执行时间小于等于当前系统时间的时候，存到List中
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            this.redisCacheService.lLeftPush(ScheduleConstants.TOPIC + keySuffix, JSON.toJSONString(task));
        }

        //2.2 当执行时间小于等于当前系统时间+五分钟 存到zset中
        //获取当前系统时间+5
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);

        if (task.getExecuteTime() <= calendar.getTimeInMillis()) {
            this.redisCacheService.zAdd(ScheduleConstants.FUTURE + keySuffix, JSON.toJSONString(task), task.getExecuteTime());
        }
    }

    /**
     * 任务保存到数据库
     *
     * @param task 任务
     * @author DiXiaoNie
     * @date 2022/12/24
     */
    private Boolean saveTaskToDB(Task task) {
        //1.新增任务到数据库表中，taskinfo和taskinfo_logs
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        this.taskinfoMapper.insert(taskinfo);

        //
        task.setTaskId(taskinfo.getTaskId());

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        taskinfoLogs.setVersion(1);
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        int i = this.taskinfoLogsMapper.insert(taskinfoLogs);
        return i == 1;
    }
}
